package com.yy.service;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.curator.framework.recipes.locks.Lease;
import org.apache.curator.framework.recipes.shared.SharedCount;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Author: chen
 * @Date: 2022/9/24 19:01
 * @Desc: 注意: 每new一个InterProcessMutex就是一把锁,重入应该使用一个锁对象
 */
@Service
public class CuratorStockService {
    @Autowired
    CuratorFramework curatorFramework;
    @Autowired
    StringRedisTemplate redisTemplate;

    InterProcessReadWriteLock readWriteLock;

    @PostConstruct
    public void init() {
        readWriteLock = new InterProcessReadWriteLock(curatorFramework, "/curatorReadWrite");
    }

    public void interProcessMutex() {
        // 获取一个封装好的基于zookeeper的可重入锁
        InterProcessMutex mutex = new InterProcessMutex(curatorFramework, "/curatorLocks");

        try {
            mutex.acquire();
            String stock = redisTemplate.opsForValue().get("stock");

            if (null != stock && stock.length() > 0) {
                int count = Integer.parseInt(stock);
                if (count > 0) {
                    redisTemplate.opsForValue().set("stock", String.valueOf(--count));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                mutex.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * maxLeases 最大租约,也就是控制资源访问的最大线程数量
     * InterProcessMutex 默认是1,等于是一个不可重入锁
     */
    public void interProcessSemaphoreV2() {
        // 获取一个封装好的基于zookeeper的不可重入锁
        InterProcessSemaphoreV2 interProcessSemaphoreV2 = new InterProcessSemaphoreV2(curatorFramework, "/curatorSemaphoreV2", 3);
        Lease lease = null;

        try {
            lease = interProcessSemaphoreV2.acquire();
            redisTemplate.opsForList().rightPush(
                    "semaphoreLog",
                    Thread.currentThread().getName() + ": 获取到了任务资源,开始执行业务逻辑"
            );

            TimeUnit.SECONDS.sleep(new Random().nextInt(10) + 10);

            redisTemplate.opsForList().rightPush(
                    "semaphoreLog",
                    Thread.currentThread().getName() + ": 处理完了业务逻辑,开始释放资源"
            );
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            interProcessSemaphoreV2.returnLease(lease);
        }
    }

    /**
     * 读锁
     */
    public void interProcessReadLock() {

        try {
            readWriteLock.readLock().acquire(10, TimeUnit.SECONDS);

            System.out.println("执行读取操作-------->");

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                // readWriteLock.readLock().release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 写锁
     */
    public void interProcessWriteLock() {

        try {
            readWriteLock.writeLock().acquire(10, TimeUnit.SECONDS);

            System.out.println("执行写入操作-------->");

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                // readWriteLock.writeLock().release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 共享计数器,类似分布式atomic
     * 可以添加监听器,当值发生改变触发回调
     * 还有 DistributedAtomicInteger 和 DistributedAtomicLong
     * 功能也差不多,可以借助这些实现countDownLatch类似的功能
     */
    public void shareCount() {

        SharedCount sharedCount = new SharedCount(curatorFramework, "/curatorShareCount", 100);

        try {
            sharedCount.start();

            int count = sharedCount.getCount();
            int random = new Random().nextInt(1000);
            sharedCount.setCount(random);

            System.out.printf("共享计数器的初始值: %s\t修改后的值: %s\n", count, random);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                sharedCount.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
